package com.whcoding.task.xxl.job.task.base;


import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.whcoding.task.xxl.job.task.utils.ThreadSettingUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

/**
 * 任务协调者 负责从Tasker取任务
 *
 * @author ZhouWenHao
 * @description 任务协调者 负责从Tasker取任务
 * @date 2023/04/04 9:15
 */
public class Coordinator<T> {
    private static final Logger logger = LoggerFactory.getLogger(Coordinator.class);

    private BaseTasker<T> tasker;
    private BaseTasker.Setting setting;
    /**
     * 协调者当前的工作状态 默认不工作
     */
    private boolean working = false;
    private DelayControl<T> delayControl;

    Coordinator(BaseTasker tasker) {
        this.tasker = tasker;
        this.setting = tasker.getSetting();
        this.delayControl = new DelayControl<T>(tasker);
    }

    /**
     * 协调者开始工作
     *
     * @param
     * @return void
     * @description 协调者开始工作
     * @author ZhouWenHao
     * @date 2023/04/04 9:15
     */
    public void start() {
        if (working) {
            return;
        }
        working = true;
        ThreadSettingUtils.getSingleThreadExecutor("coordinator-pool-%d").execute(() -> {
            while (true) {
                try {
                    assign();
                    TimeUnit.MILLISECONDS.sleep(setting.getCoordinatorSleep());
                } catch (Exception e) {
                    logger.error("coordinator loop error : {}", e.getMessage());
                }
            }
        });
    }

    /**
     * 任务指派
     *
     * @param
     * @return void
     * @description 任务指派
     * @author ZhouWenHao
     * @date 2023/04/04 9:15
     */
    private void assign() {
        if (!tasker.getCacheQueue().isEmpty()) {
            //拿到当前池子里的全部任务
            List<T> tasks = getTask();
            //按照任务数量和线程池数量分组
            List<List<T>> taskGroups = grouping(tasks);
            //向线程池推任务
            pushTask(taskGroups);
        }
        //从重试队列中查询可执行的任务并执行
        pushRetryTask(delayControl.get());
    }

    /**
     * 从tasker的cacheQueue中获取全部任务
     *
     * @param
     * @return
     * @description 从tasker的cacheQueue中获取全部任务
     * @author ZhouWenHao
     * @date 2023/04/04 9:15
     */
    private List<T> getTask() {
        List<T> tasks = Lists.newArrayList();
        ConcurrentLinkedQueue<T> cacheQueue = tasker.getCacheQueue();
        /**
         * 划重点，cacheQueue.size()不能写在for循环里，必须拿出来
         */
        int size = cacheQueue.size();
        for (int i = 0; i < size; i++) {
            T poll = cacheQueue.poll();
            if (null != poll) {
                tasks.add(poll);
            }
        }
        return tasks;
    }

    /**
     * 按照配置分组
     *
     * @param tasks 缓存队列中的全部任务
     * @return java.util.List
     * @description 按照配置分组
     * @author ZhouWenHao
     * @date 2023/04/04 9:15
     */
    private List<List<T>> grouping(List<T> tasks) {
        int taskSize = tasks.size();
        int groupSize = taskSize / setting.getCorePoolSize();
        groupSize = Math.max(groupSize, setting.getMinTaskGroupSize());
        groupSize = Math.min(groupSize, setting.getMaxTaskGroupSize());
        return Lists.partition(tasks, groupSize);
    }

    /**
     * 向线程池推任务
     *
     * @param taskGroups 任务组
     * @return void
     * @description 向线程池推任务
     * @author whcoding
     * @date 2023/4/06 9:10
     */
    private void pushTask(List<List<T>> taskGroups) {
        for (List<T> taskGroup : taskGroups) {
            try {
                pushTaskHandler(taskGroup);
            } catch (Exception e) {
                logger.error("TASKER pushTask executor error, e=[{}], data=[{}]", e.getMessage(), JSON.toJSONString(taskGroup));
            }
        }
    }

    private void pushTaskHandler(List<T> taskGroup) {
        tasker.getExecutor().execute(() -> {
            Long start = System.currentTimeMillis();
            try {
                tasker.taskRunnable(taskGroup);
                tasker.getMonitor().taskFinishNotice(true, System.currentTimeMillis() - start);
            } catch (Exception e) {
                logger.error("TASKER pushTask失败, e=[{}], data=[{}]", e.getMessage());
                //加入重试
                delayControl.add(taskGroup, e);
                tasker.getMonitor().taskFinishNotice(false, System.currentTimeMillis() - start, e);
            }
        });
    }


    /**
     * 向线程池推重试任务
     *
     * @param delayList 重试队列
     * @return void
     * @description 向线程池推重试任务
     * @author whcoding
     * @date 2023/4/06 9:10
     */
    private void pushRetryTask(List<List<T>> delayList) {
        for (List<T> taskGroup : delayList) {
            try {
                pushRetryTaskHandler(taskGroup);
            } catch (Exception e) {
                logger.error("TASKER pushRetryTask executor error, e=[{}], data=[{}]", e.getMessage());
            }
        }
    }

    private void pushRetryTaskHandler(List<T> taskGroup) {
        tasker.getExecutor().execute(() -> {
            Long start = System.currentTimeMillis();
            try {
                tasker.taskRunnable(taskGroup);
                tasker.getMonitor().taskFinishNotice(true, System.currentTimeMillis() - start, delayControl.getRetryTimes(taskGroup));
                //重试成功移除延迟中心的任务
                delayControl.remove(taskGroup);
            } catch (Exception e) {
                logger.error("TASKER pushRetryTask失败, e=[{}], data=[{}]", e.getMessage(), JSON.toJSONString(taskGroup));
                //加入重试
                delayControl.add(taskGroup, e);
                tasker.getMonitor().taskFinishNotice(false, System.currentTimeMillis() - start, delayControl.getRetryTimes(taskGroup), e);
            }
        });
    }

}
